package com.shujia.dwi

import com.shujia.common.utils.{Geography, SparkMain}
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DwiResRegnMergelocationMskDay extends SparkMain {

  override def run(spark: SparkSession): Unit = {

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 加载oidd的数据
    val oiddDF: DataFrame = spark.table("ods.ods_oidd").where($"day_id" === dayId)


    // 1、将时间字段切开来，得到起始时间及结束时间 对同一个用户的所有位置数据按照起始时间进行升序排序
    val filterDF: DataFrame = oiddDF
      .withColumn("start_t", split($"start_time", ",")(1))
      .withColumn("end_t", split($"start_time", ",")(0))

      // 在时间轴上进行聚类：解决相同网格内连续数据之间有时间交叠的问题
      // 使用lag窗口函数 取上一条数据的网格id
      .withColumn("last_grid_id", lag("grid_id", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 对比当前数据的网格id和上一条数据的网格id是否相同，相同则置0，不同置为1，得到flag列
      .withColumn("flag", when($"grid_id" === $"last_grid_id", 0).otherwise(1))
      // 使用sum基于窗口对flag列进行累加求和 得到group列
      .withColumn("group", sum($"flag") over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 对同一组内的数据进行合并
      .groupBy($"group", $"mdn", $"grid_id", $"bsid", $"longi", $"lati", $"county_id")
      .agg(min($"start_t") as "min_start_t", max($"end_t") as "max_end_t")

      // 使用当前数据的开始时间减去上一条数据的开始时间 作为相邻两条数据的时间差：不同网格内相邻记录有时间交叠问题
      .withColumn("last_min_start_t", lag("min_start_t", 1) over Window.partitionBy($"mdn").orderBy($"min_start_t"))
      // 计算时间差
      .withColumn("diff_time", unix_timestamp($"min_start_t", "yyyyMMddHHmmss") - unix_timestamp($"last_min_start_t", "yyyyMMddHHmmss"))

      // 使用lag窗口函数 基于排序后的记录 向前取上一条数据中的经纬度
      .withColumn("last_longi", lag("longi", 1) over Window.partitionBy($"mdn").orderBy($"min_start_t"))
      .withColumn("last_lati", lag("lati", 1) over Window.partitionBy($"mdn").orderBy($"min_start_t"))
      // 当处理第一条数据时 无法取到上一条的数据 值就会出现null的情况
      // 当值为null则直接将距离置为0 否则就计算两个经纬度之间的距离
      .withColumn("distance", when($"last_longi".isNull, 0).otherwise(calculateLengthWithLongiLati($"last_longi", $"last_lati", $"longi", $"lati")))
      // 计算用户移动的速度(单位：米/秒)
      .withColumn("speed", round($"distance" / $"diff_time", 2))

      /**
       * 常见交通工具的速度：
       * 步行：1~2m/s
       * 跑步：3~4m/s
       * 自行车：5~10m/s
       * 电动车：5~16m/s
       * 摩托车：5~28m/s
       * 自驾：10~35m/s
       * 火车：30m/s左右
       * 高铁：100m/s左右
       * 飞机：200~300m/s
       */
      // 过滤掉速度超过300m/s的数据
      .where($"speed" < 300)
      // 整理数据
      .select(
        upper(md5(concat($"mdn", expr("'shujia'")))) // 对手机号进行脱敏
        , $"min_start_t" as "start_date"
        , $"max_end_t" as "end_date"
        , $"county_id"
        , $"longi"
        , $"lati"
        , $"bsid"
        , $"grid_id"
      )

    // 保存数据
    filterDF
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dwi/dwi_res_regn_mergelocation_msk_d/day_id=$dayId")

    // 增加分区
    spark.sql(
      s"""
         |alter table dwi.dwi_res_regn_mergelocation_msk_d add if not exists partition(day_id=$dayId)
         |""".stripMargin)
  }

  /**
   * 提交命令：
   * spark-submit --master yarn-client --conf spark.sql.shuffle.partitions=10 --class com.shujia.dwi.DwiResRegnMergelocationMskDay --jars common-1.0.jar dwi-1.0.jar 20230105
   */

}
